package com.icehand.redis.bloom;

import com.google.common.hash.Funnels;
import com.google.common.hash.Hashing;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

import java.nio.charset.Charset;
import java.util.List;

/**
 * redis 实现的bloom filter
 * @author icehand
 */
public class RedisBloomFilter {
    private final RedisTemplate redisTemplate;

    /**
     * 预计插入量
     */
    private long expectedInsertions;

    /**
     * 可接受的错误率
     */
    private double fpp;

    private String redisKeyPrefix;

    /**
     * bit数组长度
     */
    private long numBits;
    /**
     * hash函数数量
     */
    private int numHashFunctions;

    public RedisBloomFilter(RedisTemplate redisTemplate, long expectedInsertions, double fpp, String redisKeyPrefix){
        this.redisKeyPrefix = redisKeyPrefix;
        this.redisTemplate = redisTemplate;
        this.expectedInsertions = expectedInsertions;
        this.fpp = fpp;
        this.numBits = optimalNumOfBits(expectedInsertions, fpp);
        this.numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);
    }

    public RedisBloomFilter(RedisTemplate redisTemplate, long expectedInsertions, double fpp){
        this(redisTemplate,expectedInsertions,fpp,"bf:");
    }

    public RedisBloomFilter(RedisTemplate redisTemplate, String redisKeyPrefix){
        this(redisTemplate,50000000L,0.001F,redisKeyPrefix);
    }

    public RedisBloomFilter(RedisTemplate redisTemplate) {
        this(redisTemplate,"bf:");
    }

    public void setExpectedInsertions(long expectedInsertions) {
        this.expectedInsertions = expectedInsertions;
    }

    public void setFpp(double fpp) {
        this.fpp = fpp;
    }

    public void setRedisKeyPrefix(String redisKeyPrefix) {
        this.redisKeyPrefix = redisKeyPrefix;
    }

    /**
     * 计算hash函数个数
     */
    private int optimalNumOfHashFunctions(long n, long m) {
        return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
    }

    /**
     * 计算bit数组长度
     */
    private long optimalNumOfBits(long n, double p) {
        if (p == 0) {
            p = Double.MIN_VALUE;
        }
        return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
    }

    /**
     * 判断keys是否存在于集合
     */
    public boolean isExist(String where, String key) {
        long[] indexes = getIndexes(key);

        List list = redisTemplate.executePipelined((RedisCallback<Boolean>) redisConnection -> {
            redisConnection.openPipeline();
            for (long index : indexes) {
                redisConnection.getBit(getRedisKey(where).getBytes(), index);
            }
            return null;
        });

        if(list.size() == 0){
            return false;
        }

        return !list.contains(Boolean.FALSE);
    }

    /**
     * 将key存入redis bitmap
     */
    public void put(String where, String key) {
        long[] indexs = getIndexes(key);
        redisTemplate.executePipelined((RedisCallback<Object>) redisConnection -> {
            for (long index : indexs) {
                redisConnection.setBit(getRedisKey(where).getBytes(), index, true);
            }
            return null;
        });
    }

    /**
     * 根据key获取bitmap下标
     */
    private long[] getIndexes(String key) {
        long hash1 = hash(key);
        long hash2 = hash1 >>> 16;
        long[] result = new long[numHashFunctions];
        for (int i = 0; i < numHashFunctions; i++) {
            long combinedHash = hash1 + i * hash2;
            if (combinedHash < 0) {
                combinedHash = ~combinedHash;
            }
            result[i] = combinedHash % numBits;
        }
        return result;
    }

    /**
     * 获取一个hash值
     */
    private long hash(String key) {
        Charset charset = Charset.forName("UTF-8");
        return Hashing.murmur3_128().hashObject(key, Funnels.stringFunnel(charset)).asLong();
    }

    private String getRedisKey(String where) {
        return redisKeyPrefix + where;
    }
}
